package com.example.springbootkafka.kafka;

import com.alibaba.fastjson.JSON;
import com.example.springbootkafka.constants.BaseConstants;
import com.example.springbootkafka.dto.TtlMsgDto;
import com.example.springbootkafka.service.ProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.SerializationUtils;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class Consumer {

    @Autowired
    ProducerService producerService;


    @KafkaListener(topics = {"${spring.kafka.consumer.topic}"})
    public void topicTest(ConsumerRecord<String, String> record) {
        String msgBody = record.value();
//        log.info("接收到数据，topic = {}，value = {}", record.topic(), msgBody);
        try {
            Long ttl = -1L;
            String deadTopic = "";
            // 模拟续约的情况
            for (Header header : record.headers()) {
                if (header.key().equals(BaseConstants.TTL_KEY)) {
                    ttl = (Long) SerializationUtils.deserialize(header.value());
                }
                if (header.key().equals(BaseConstants.DEAD_TOPIC_KEY)) {
                    deadTopic = (String) SerializationUtils.deserialize(header.value());
                }
            }

            // ttl为非 -1 的，说明是延时消息
            consumerDelayedMsgV000(record, msgBody, ttl, deadTopic);

            //throw new RuntimeException("手动模拟异常");
        } catch (Exception e) {
            log.error("处理kafka数据出错，value = {},错误信息 = {}", msgBody, e);
            //报错时处理逻辑
        } finally {
            // 手动提交offset
        }


    }

    /**
     * 有问题的消息处理：还原生产上面的问题
     */
    private void consumerDelayedMsgV0(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) {
        if (ttl.compareTo(-1L) != 0) {

        }

        TtlMsgDto msgBodyJo = JSON.parseObject(msgBody, TtlMsgDto.class);
        LocalDateTime expireTime = msgBodyJo.getExpireTime();
        Duration duration = Duration.between(LocalDateTime.now(), expireTime);
        long srcTtl = duration.toMillis();
        if (srcTtl <= 1000) {
            log.error("延时消息小于等于1秒，立即消费=====>");
        } else {
            log.info("延时消息大于1秒，重回队列");
            if (srcTtl > Integer.MAX_VALUE) {
                srcTtl = Long.valueOf(Integer.MAX_VALUE);
            }
            // 注意：record.key() 是消息的唯一键
            producerService.sendTtlMessage(record.topic(), msgBody, srcTtl);
        }
    }

    /**
     * 尽可能 解决 原生产上面的问题
     */
    private void consumerDelayedMsgV00(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) {

        TtlMsgDto msgBodyJo = JSON.parseObject(msgBody, TtlMsgDto.class);
        LocalDateTime expireTime = msgBodyJo.getExpireTime();
        Duration duration = Duration.between(LocalDateTime.now(), expireTime);
        long srcTtl = duration.toMillis();
        if (ttl > -1 && ttl <= BaseConstants.MAX_TTL) {
            if (srcTtl <= 1000) {
                log.error("todo-真正的处理业务消息，record = {}, ttl = {}, deadTopic = {}", msgBody, ttl, deadTopic);
                //log.error("延时消息小于等于1秒，立即消费=====>");
            } else {
//                log.info("延时消息大于1秒，重回队列");
                // 注意：record.key() 是消息的唯一键
                producerService.sendTtlMessage(record.topic(), msgBody, record.key(), srcTtl);
            }
        } else if (StringUtils.isEmpty(deadTopic)) {
            log.error("延时消息首次大于支持的最大延时时间 {} 秒，重回队列", BaseConstants.MAX_TTL / 1000);
            // 注意：record.key() 是消息的唯一键
            producerService.sendTtlMessage(record.topic(), msgBody, record.key(), BaseConstants.MAX_TTL);
        }


    }

    /**
     * 尽可能 解决 原生产上面的问题
     */
    private void consumerDelayedMsgV000(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) throws InterruptedException {

        TtlMsgDto msgBodyJo = JSON.parseObject(msgBody, TtlMsgDto.class);
        LocalDateTime expireTime = msgBodyJo.getExpireTime();
        Duration duration = Duration.between(LocalDateTime.now(), expireTime);
        long srcTtl = duration.toMillis();
        long numCount = srcTtl / BaseConstants.MAX_TTL;

        // 给延时消息-续时的关键-
        // 1.超出最大延时时间n倍后，需要给新的key = key:n
        // 2.业务端判断消息未达到延时后，需要延时1秒后，再入队列
        // 3.超出最大延时时间，在消息拦截器这里：应该借鉴主流mq(rabbitmq) 立马消费消息
        if (ttl > -1 && ttl <= BaseConstants.MAX_TTL) {
            realHanleDelayedMsg(record, msgBody, ttl, deadTopic, srcTtl, numCount);
        } else if (StringUtils.isEmpty(deadTopic)) {
            if (srcTtl > BaseConstants.MAX_TTL) {
                log.error("延时消息 是 支持的最大延时时间 {} 秒 的 {} 倍，重回队列", BaseConstants.MAX_TTL / 1000, numCount);
                producerService.sendTtlMessage(record.topic(), msgBody, record.key() + ":" + numCount, BaseConstants.MAX_TTL);
                return;
            }
            realHanleDelayedMsg(record, msgBody, ttl, deadTopic, srcTtl, numCount);
        }


    }

    private void realHanleDelayedMsg(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic, long srcTtl, long numCount) throws InterruptedException {
        if (srcTtl <= 1000) {
            log.error("todo-真正的处理业务消息，record = {}, ttl = {}, deadTopic = {}", msgBody, ttl, deadTopic);
        } else {
            log.info("延时消息大于1秒，重回队列 srcTtl = {} record = {}, ttl = {}, deadTopic = {}", srcTtl, msgBody, ttl, deadTopic);
            TimeUnit.SECONDS.sleep(1L);


            if (srcTtl >= BaseConstants.MAX_TTL) {
                log.error("延时消息 是 支持的最大延时时间 {} 秒 的 {} 倍，忽略消息", BaseConstants.MAX_TTL / 1000, numCount);
                return;
            }

            // 注意：record.key() 是消息的唯一键
            producerService.sendTtlMessage(record.topic(), msgBody, record.key(), srcTtl);
        }
    }

    /**
     * 正确的消息处理：
     */
    private void consumerDelayedMsgV1(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) {

        // 判断真正的延时消息（即：已经移除了 延时特性的消息，在timeWheel中 移除了ttl）
        if (ttl.compareTo(-1L) == 0) {
            log.info("todo-真正的处理业务消息，record = {}, ttl = {}, deadTopic = {}", msgBody, ttl, deadTopic);
            // todo: hcj 业务上的续期有问题
            //bizHandleOverMaxDelayedTime(record, msgBody, ttl, record.topic());
        } else {
            log.info(" ttl = {}", ttl);
        }
    }

    /**
     * 业务处理超时（即：延时时间 超过 延时消息支持的最大延时时间）
     */
    private boolean bizHandleOverMaxDelayedTime(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) {
        TtlMsgDto msgBodyJo = JSON.parseObject(msgBody, TtlMsgDto.class);
        LocalDateTime expireTime = msgBodyJo.getExpireTime();
        Duration duration = Duration.between(LocalDateTime.now(), expireTime);
        long srcTtl = duration.toMillis();
        if (srcTtl > BaseConstants.MAX_TTL) {
            log.info("延时消息大于支持的最大延时时间，重回队列 ttl = {} srcTtl = {}", ttl, srcTtl);
            srcTtl = srcTtl - BaseConstants.MAX_TTL;
            log.info("超出延时支持的最大时间，新的延时时间- srcTtl = {}", srcTtl);
            producerService.sendTtlMessage(deadTopic, msgBody, record.key(), srcTtl);
            return true;
        } else if (srcTtl == BaseConstants.MAX_TTL) {
            producerService.sendTtlMessage(deadTopic, msgBody, record.key(), srcTtl);
            return true;
        }

        return false;
    }

    /**
     * 正确的消息处理：
     */
    private void consumerDelayedMsgV3(ConsumerRecord<String, String> record, String msgBody, Long ttl, String deadTopic) {

        long diffTime = System.currentTimeMillis() - record.timestamp();
        if (diffTime >= ttl) {
            log.info("todo-真正的处理业务消息，record = {}, ttl = {}, deadTopic = {}", msgBody, ttl, deadTopic);
        } else {
            log.info("未到消费时间，重回队列 - diffTime = {} ttl = {}", diffTime, ttl);
            //producerService.sendTtlMessage(record.topic(), record.value(), record.key(), ttl);
        }
    }

}
